/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "xdrbase.h"
#include "limit.h"
#include "jm.h"

float  exchIntvl = EXCHINTVL;       /* 15 */
short  gintInactivityLimit = HOSTINACTIVITYLIMIT; /* 5 */
short  gintMasterInactLimit = MASTERINACTIVITYLIMIT;  /* 2 */
short  gintInactiveRetryLimit = INACTIVITYRETRYLIMIT;

static u_int sintSeqNo;
static time_t gtimeMasterJmBoot;

static int send_static_info(void);

/**
 * @brief       Tell JS that my running job status or job resource usage is changed.
 *
 * @param[in]   reqcode      #1: BATCH_STATUS_JOB or BATCH_RUSAGE_JOB
 * @param[in]   job          #2: my running job entity
 * @param[in]   status       #3: new status of the job
 * @param[in]   error        #4: if jm reboot or not
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 */
int
status_job(js_request_t reqcode, struct jobEntry *job, int status, unsigned int error)
{
    if (0 > gintJsFd) { // jm is not ready for sending status to js.
        return (-1);
    }

    if (status == JOB_STATUS_EXIT) {
        job->succeed = FALSE;
    }

    if (MASK_STATUS(status) == JOB_STATUS_DONE) {
        job->succeed = TRUE;
    }

    if (IS_JOB_POST_DONE(status) || IS_JOB_POST_ERR(status)) {
        if (!job->succeed) {
            return 0;
        }
    }

    if (NULL == gstrBossNode) {
        return -1;
    }

    if (0 > job->comm_failcnt) {
        job->comm_failcnt = -INFINIT_INT;
        return 0;
    }

    unsigned int i;
    if (IS_JOB_ENDED(status)) {

        for (i=0; i<job->job_base.number_node; i++) {
            if (job->max_usage[i].mem > job->inst_usage[i].mem || 0 == job->inst_usage[i].mem) {
                job->inst_usage[i].mem = job->max_usage[i].mem;
            }

            if (job->max_usage[i].swap > job->inst_usage[i].swap || 0 == job->inst_usage[i].swap) {
                job->inst_usage[i].swap = job->max_usage[i].swap;
            }

            if (job->max_usage[i].stime > job->inst_usage[i].stime || 0 == job->inst_usage[i].stime) {
                job->inst_usage[i].stime = job->max_usage[i].stime;
            }

            if (job->max_usage[i].utime > job->inst_usage[i].utime || 0 == job->inst_usage[i].utime) {
                job->inst_usage[i].utime = job->max_usage[i].utime;
            }
        }
    }

    struct request req;
    memset(&req, 0, sizeof(struct request));

    unsigned int num = 4; // jobid, job_pid, job_pgid, status

    if (0 < job->job_base.act_pid) {
        num ++;
    }

    if (0 < job->job_base.reasons) {
        num ++;
    }

    if (0 < job->job_base.subreasons) {
        num ++;
    }

    if (0 < error) {
        num ++;
    }

    num ++; // execute_uid;

    if (0 != job->w_status) {
        num ++;
    } 

    if (NULL != job->job_base.execute_cwd && '\0' != job->job_base.execute_cwd[0]) {
        num ++;
    }

    if (NULL != job->job_base.execute_home && '\0' != job->job_base.execute_home[0]) {
        num ++;
    }

    if (NULL != job->job_base.execute_user && '\0' != job->job_base.execute_user[0]) {
        num ++;
    }

    num ++; // number_node
    for (i=0; i<job->job_base.number_node; i++) {

        num ++; // mem

        if (0 < job->inst_usage[i].swap) {
            num ++;
        }

        if (0 < job->inst_usage[i].utime) {
            num ++;
        }

        if (0 < job->inst_usage[i].stime) {
            num ++;
        }

        if (0 < job->inst_usage[i].npids) {
            num ++;
            num += job->inst_usage[i].npids;
        }

        if (0 < job->inst_usage[i].npgids) {
            num ++;
            num += job->inst_usage[i].npgids;
        }
    }

    if (0 != job->actcmd_state) {
        num ++;
    } 

    if (0 != job->job_base.act_signal) {
        num ++;
    } 

    req.number = num;
    req.keyvalues = (struct keyvalue *)calloc(num, sizeof(struct keyvalue));
    if (NULL == req.keyvalues) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", num*sizeof(struct keyvalue));
        relaunch();
    }

    num = 0;

    req.keyvalues[num].key = REQUEST_KEY_JOBIDS;
    req.keyvalues[num].value = (char *)calloc(32, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 32);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 32, "%lld", job->job_base.jobid);
    num ++;

    req.keyvalues[num].key = REQUEST_KEY_JOBPID;
    req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.job_pid);
    num ++;

    req.keyvalues[num].key = REQUEST_KEY_JOBPGID;
    req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.job_pgid);
    num ++;

    req.keyvalues[num].key = REQUEST_KEY_JOBSTATUS;
    req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 12, "%d", status);
    num ++;

    if (0 < job->job_base.act_pid) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_ACTPID;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.act_pid);
        num ++;
    }

    if (0 < job->job_base.reasons) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_REASON;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.reasons);
        num ++;
    }

    if (0 < job->job_base.subreasons) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_SUBREASON;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.subreasons);
        num ++;
    }

    if (0 < error) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_ERROR;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", error);
        num ++;
    }

    req.keyvalues[num].key = REQUEST_KEY_EXECUTE_UID;
    req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.execute_uid);
    num ++;

    if (0 != job->w_status) {
        req.keyvalues[num].key = REQUEST_KEY_EXITSTATUS;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->w_status);
        num ++;
    } 

    if (NULL != job->job_base.execute_cwd && '\0' != job->job_base.execute_cwd[0]) {
        req.keyvalues[num].key = REQUEST_KEY_EXECUTE_CWD;
        req.keyvalues[num].value = copy_string(job->job_base.execute_cwd);
        num ++;
    }

    if (NULL != job->job_base.execute_home && '\0' != job->job_base.execute_home[0]) {
        req.keyvalues[num].key = REQUEST_KEY_EXECUTE_HOME;
        req.keyvalues[num].value = copy_string(job->job_base.execute_home);
        num ++;
    }

    if (NULL != job->job_base.execute_user && '\0' != job->job_base.execute_user[0]) {
        req.keyvalues[num].key = REQUEST_KEY_USERNAME;
        req.keyvalues[num].value = copy_string(job->job_base.execute_user);
        num ++;
    }

    req.keyvalues[num].key = REQUEST_KEY_EXECUTE_NUMNODE;
    req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == req.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.number_node);
    num ++;

    for (i=0; i<job->job_base.number_node; i++) {

        req.keyvalues[num].key = REQUEST_KEY_USAGE_MEM;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%ld", job->inst_usage[i].mem);
        num ++;

        if (0 < job->inst_usage[i].swap) {
            req.keyvalues[num].key = REQUEST_KEY_USAGE_SWAP;
            req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == req.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(req.keyvalues[num].value, 12, "%ld", job->inst_usage[i].swap);
            num ++;
        }

        if (0 < job->inst_usage[i].utime) {
            req.keyvalues[num].key = REQUEST_KEY_USAGE_UTIME;
            req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == req.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(req.keyvalues[num].value, 12, "%ld", job->inst_usage[i].utime);
            num ++;
        }

        if (0 < job->inst_usage[i].stime) {
            req.keyvalues[num].key = REQUEST_KEY_USAGE_STIME;
            req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == req.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(req.keyvalues[num].value, 12, "%ld", job->inst_usage[i].stime);
            num ++;
        }

        if (0 < job->inst_usage[i].npids) {
            req.keyvalues[num].key = REQUEST_KEY_USAGE_NPID;
            req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == req.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(req.keyvalues[num].value, 12, "%d", job->inst_usage[i].npids);
            num ++;

            unsigned int j;
            for (j=0; j<job->inst_usage[i].npids; j++) {
                req.keyvalues[num].key = REQUEST_KEY_USAGE_PIDINFO;
                req.keyvalues[num].value = (char *)calloc(36, sizeof(char));
                if (NULL == req.keyvalues[num].value) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 36);
                    relaunch();
                }

                snprintf(req.keyvalues[num].value, 36, "%d %d %d",
                         job->inst_usage[i].pidinfo[j].pid, job->inst_usage[i].pidinfo[j].ppid,
                         job->inst_usage[i].pidinfo[j].pgid);
                num ++;
            }
        }

        if (0 < job->inst_usage[i].npgids) {
            req.keyvalues[num].key = REQUEST_KEY_USAGE_NPGID;
            req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == req.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(req.keyvalues[num].value, 12, "%d", job->inst_usage[i].npgids);
            num ++;

            unsigned int j;
            for (j=0; j<job->inst_usage[i].npgids; j++) {
                req.keyvalues[num].key = REQUEST_KEY_USAGE_PGID;
                req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
                if (NULL == req.keyvalues[num].value) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                    relaunch();
                }
                snprintf(req.keyvalues[num].value, 12, "%d", job->inst_usage[i].pgid[j]);
                num ++;
            }
        }
    }

    if (0 != job->actcmd_state) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_ACTSTATE;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->actcmd_state);
        num ++;
    } 

    if (0 != job->job_base.act_signal) {
        req.keyvalues[num].key = REQUEST_KEY_JOB_ACTSIGNAL;
        req.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == req.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(req.keyvalues[num].value, 12, "%d", job->job_base.act_signal);
        num ++;
    } 
 
    unsigned int size;
    size = gf_xdrsize_request(&req);
    size += HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        gf_xdr_free(gf_xdr_request, &req);
        relaunch();
    }

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    struct header hdr;
    gf_init_header(&hdr);
    hdr.reqcode = reqcode;

    char strJobId[32];
    if (!gf_xdr_message(&xdrs, &req, &hdr, NULL, gf_xdr_request, NULL)) {

        log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M,
                    __func__, jobid_2string(job->job_base.jobid, strJobId), "gf_xdr_request");
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_request, &req);
        relaunch();
    }

    int cc;
    cc = gf_callslave(gstrBossNode, g_js_port, g_conntimeout, g_recvtimeout,
                      XDR_GETPOS(&xdrs), request, NULL, NULL, &gintJsFd);
    if (0 > cc) {

        log_message(ERR, NOCLASS, "%s: Failed to reach js on host <%s> for job <%s>",
                    __func__, gstrBossNode, jobid_2string(job->job_base.jobid, strJobId));
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_chan_close(gintJsFd);
        gintJsFd = -1;
        gf_xdr_free(gf_xdr_request, &req);

        if (gboolMaster) {
            gboolJSReady = FALSE;
        }

        if (-2 == cc) {
            FREEUP(gstrBossNode);
        }

        return -1;
    } else if (0 == cc) {
        job->update_usage = FALSE;
        job->report_time = time(NULL);
    }

    xdr_destroy(&xdrs);
    FREEUP(request);
    gf_xdr_free(gf_xdr_request, &req);
    
    return 0;
} // end function status_job

/**
 * @brief       Tell JS that I am starting.
 *
 * @note        Then I will get my running jobs from JS.
 */
void
notify_startup(void)
{
    char request[HEADER_LEN+10];
    XDR xdrs;
    struct header tohdr;

    /* manager node must have been set when calloing master. */
    if (NULL == gstrBossNode) {
        return;
    }

    /* It is possible that I'm master now but js is not started. */
    if (0 < gintJsFd && strcmp(gstrBossNode, gptrMyHost->name) == 0) {
        return;
    }

    xdrmem_create(&xdrs, request, HEADER_LEN+10, XDR_ENCODE);

    gf_init_header(&tohdr);
    tohdr.reqcode = BATCH_SLAVE_RESTART;

    if (!gf_xdr_message(&xdrs, NULL, &tohdr, NULL, NULL, NULL)) {
        log_message(ERR, NOCLASS, "%s: failed to encode BATCH_SLAVE_RESTART.", __func__);
        xdr_destroy(&xdrs);
        jm_die(DAEMON_QUIT_FATAL);
    }

    int cc, fd = -1;
    int connTimeout = 1;
    cc = gf_callmaster(gstrBossNode, g_js_port, connTimeout, g_recvtimeout, 1,
                       XDR_GETPOS(&xdrs), request, NULL, NULL, &fd);
    if (cc >= 0 && fd > 0) {

        gintJsFd = fd;
        xdr_destroy(&xdrs);
        gf_chan_state(fd, CHANNEL_LONGCONN);

        return;
    }

    xdr_destroy(&xdrs);

    return;
} // end function notify_startup

/**
 * @brief       Tell JM my executing job status(pending, running).
 *
 * @param[in]   status       #1: job status
 * @param[in]   reason       #2: reason why set job pending status
 * @param[in]   job          #3: job entity
 */
void
childjm_set_jobstatus(int status, int reason, struct jobEntry *job)
{
    unsigned int num = 2; /* jobid, status */

    if (0 < reason) {
        num ++;
    }

    if (0 < job->job_base.job_pid) {
        num ++;
    }

    if (0 < job->job_base.job_pgid) {
        num ++;
    }

    if (0 <= job->execute_gid) {
        num ++;
    }

    if (0 <= job->job_base.execute_uid) {
        num ++;
    }

    if (0 < job->execute_flag) {
        num ++;
    }

    if (strlen(job->job_base.execute_user) > 0) {
        num ++;
    }

    if (strlen(job->job_base.execute_cwd) > 0) {
        num ++;
    }

    if (strlen(job->job_base.execute_home) > 0) {
        num ++;
    }

    if (!FLOAT_EQUAL_ZERO(job->cpu_time)) {
        num ++;
    }

    num ++; /* w_status */

    struct request jobRequest;
    memset(&jobRequest, 0, sizeof(struct request));

    jobRequest.number = num;
    jobRequest.keyvalues = (struct keyvalue *)calloc(num, sizeof(struct keyvalue));
    if (NULL == jobRequest.keyvalues) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", jobRequest.number*sizeof(struct keyvalue));
        relaunch();
    }

    num = 0;
    jobRequest.keyvalues[num].key = REQUEST_KEY_JOBIDS;
    jobRequest.keyvalues[num].value = (char *)calloc(32, sizeof(char));
    if (NULL == jobRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 32);
        relaunch();
    }
    jobid_2longstring(job->job_base.jobid, jobRequest.keyvalues[num].value);
    num ++;

    jobRequest.keyvalues[num].key = REQUEST_KEY_JOBSTATUS;
    jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == jobRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(jobRequest.keyvalues[num].value, 12, "%d", status);
    num ++;

    if (0 < reason) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_PENDREASON;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", reason);
        num ++;
    }

    if (0 < job->job_base.job_pid) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_JOBPID;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->job_base.job_pid);
        num ++;
    }

    if (0 < job->job_base.job_pgid) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_JOBPGID;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->job_base.job_pgid);
        num ++;
    }

    if (0 <= job->execute_gid) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_EXECUTE_GID;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->execute_gid);
        num ++;
    }

    if (0 <= job->job_base.execute_uid) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_EXECUTE_UID;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->job_base.execute_uid);
        num ++;
    }

    if (0 < job->execute_flag) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_EXECUTE_FLAG;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->execute_flag);
        num ++;
    }

    if (strlen(job->job_base.execute_user) > 0) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_USERNAME;
        jobRequest.keyvalues[num].value = copy_string(job->job_base.execute_user);
        num ++;
    }

    if (strlen(job->job_base.execute_cwd) > 0) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_EXECUTE_CWD;
        jobRequest.keyvalues[num].value = copy_string(job->job_base.execute_cwd);
        num ++;
    }

    if (strlen(job->job_base.execute_home) > 0) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_EXECUTE_HOME;
        jobRequest.keyvalues[num].value = copy_string(job->job_base.execute_home);
        num ++;
    }

    if (!FLOAT_EQUAL_ZERO(job->cpu_time)) {
        jobRequest.keyvalues[num].key = REQUEST_KEY_CPUTIME;
        jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == jobRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(jobRequest.keyvalues[num].value, 12, "%.2f", job->cpu_time);
        num ++;
    }

    jobRequest.keyvalues[num].key = REQUEST_KEY_EXITSTATUS;
    jobRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == jobRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(jobRequest.keyvalues[num].value, 12, "%d", job->w_status);
    num ++;

    unsigned int size;
    size = gf_xdrsize_request(&jobRequest);
    size += HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {
        gf_xdr_free(gf_xdr_request, &jobRequest);
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
	relaunch();
    }

    struct header hdr;
    gf_init_header(&hdr);
    hdr.reqcode = JM_JOB_SETUP;

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    char strJobId[32];
    if (!gf_xdr_message(&xdrs, &jobRequest, &hdr, NULL, gf_xdr_request, NULL)) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S,
                    __func__, jobid_2string(job->job_base.jobid, strJobId),
                    "gf_xdr_message/gf_xdr_request");
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_request, &jobRequest);
        exit(-1);
    }

    int cc, retryInterval;
    for (retryInterval = 5, cc = -1; cc != 0;) {

        int fd = -1;
        char *reply = NULL;
        cc = gf_callslave(gptrMyHost->name, g_jm_port, g_conntimeout, g_recvtimeout,
                          XDR_GETPOS(&xdrs), request, &hdr, &reply, &fd);
        if (cc < 0) {
            log_message(DEBUG, EXECUTE, "%s: gf_callslave(%s:%d) for job <%s> failed.",
                        __func__, gptrMyHost->name, g_jm_port,
                        jobid_2string(job->job_base.jobid, strJobId));
            gf_sleep(retryInterval*1000);

            retryInterval --;
            if (retryInterval == 0)
                break;
        }

        if (fd >= 0) {
            gf_chan_close(fd);
        }
    }

    gf_xdr_free(gf_xdr_request, &jobRequest);
    xdr_destroy(&xdrs);
    FREEUP(request);

    if (hdr.reqcode != 0 || cc < 0) {

        log_message(DEBUG, EXECUTE, "%s: setup job <%s> got error[%d, %d].",
                    __func__, jobid_2string(job->job_base.jobid, strJobId), hdr.reqcode, cc);
        exit(-1);
    }

    if (status & JOB_STATUS_RUN) {
        return;
    }

    exit(job->w_status);
} // end function childjm_set_jobstatus

/**
 * @brief       Send my load values to master.
 *
 * @note        send load from time to time.
 */
void
send_myload(void)
{
    if (NULL == gstrBossNode) { // do not know master for now.
        gptrMyCluster->number_inactivity ++;
        return;
    }

    struct request loadReq;
    memset(&loadReq, 0, sizeof(struct request));

    unsigned int num = 6 + gptrSharedConf->qpinfo->num_index;

    num += 1 + GET_INTNUM(gptrSharedConf->qpinfo->num_index);

    if (0 < LIST_NUM_ENTS(gptrMyHost->inst_table)) {
        num ++;
        num += 2 * LIST_NUM_ENTS(gptrMyHost->inst_table);
    }

    if (0 < gptrMyHost->num_GPU && NULL != gptrMyHost->gpuinfo_dynamic) {
        // node has GPU
        num += 5 * gptrMyHost->num_GPU;
        num++;
    }

    loadReq.keyvalues = (key_value_t *)calloc(num, sizeof(key_value_t));
    if (NULL == loadReq.keyvalues) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", num*sizeof(key_value_t));
        relaunch();
    }

    num = 0;
    loadReq.keyvalues[num].key = REQUEST_KEY_NUM_LOADINDEX;
    loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == loadReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrSharedConf->qpinfo->num_index);
    num ++;

    loadReq.keyvalues[num].key = REQUEST_KEY_NUM_USER_LOADINDEX;
    loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == loadReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrSharedConf->qpinfo->num_user_index);
    num ++;

    unsigned int i;
    for (i=0; i<(1+GET_INTNUM(gptrSharedConf->qpinfo->num_index)); i++) {
        loadReq.keyvalues[num].key = REQUEST_KEY_NODE_STATUS;
        loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == loadReq.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrMyHost->status[i]);
        num ++;
    }

    loadReq.keyvalues[num].key = REQUEST_KEY_LOAD_SEQNO;
    loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == loadReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(loadReq.keyvalues[num].value, 12, "%d", sintSeqNo++);
    num ++;

    loadReq.keyvalues[num].key = REQUEST_KEY_CLUSTER_CHECKSUM;
    loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == loadReq.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrMyCluster->checksum);
    num ++;

    for (i=0; i<gptrSharedConf->qpinfo->num_index; i++) {
        loadReq.keyvalues[num].key = REQUEST_KEY_NUM_LOAD_VALUE;
        loadReq.keyvalues[num].value = (char *)calloc(24, sizeof(char));
        if (NULL == loadReq.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 24);
            relaunch();
        }
        snprintf(loadReq.keyvalues[num].value, 24, "%f", gptrMyHost->load[i]);
        num ++;
    }

    if (0 < LIST_NUM_ENTS(gptrMyHost->inst_table)) {
        loadReq.keyvalues[num].key = REQUEST_KEY_NUM_RESOURCE;
        loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == loadReq.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(loadReq.keyvalues[num].value, 12, "%d", LIST_NUM_ENTS(gptrMyHost->inst_table));
        num ++;

        HASH_WALK_T walk;
        char *key;
        struct resourceInstance *resInst;
        gf_hash_walk_start(gptrMyHost->inst_table, &walk);
        while (NULL != (resInst=(struct resourceInstance *)gf_hash_walk(&walk, &key))) {
            loadReq.keyvalues[num].key = REQUEST_KEY_RESOURCE;
            loadReq.keyvalues[num].value = copy_string(resInst->resource->name);
            num ++;

            loadReq.keyvalues[num].key = REQUEST_KEY_RESOURCE_VALUE;
            loadReq.keyvalues[num].value = copy_string(resInst->org_value);
            num ++;
        }
    }

    if (0 < gptrMyHost->num_GPU && NULL != gptrMyHost->gpuinfo_dynamic){
        loadReq.keyvalues[num].key = REQUEST_KEY_NUM_GPU;
        loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == loadReq.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrMyHost->num_GPU);
        num ++;

        for (i = 0; i < gptrMyHost->num_GPU; i++) {
            loadReq.keyvalues[num].key = REQUEST_KEY_GPU_MEMORY;
            loadReq.keyvalues[num].value = (char *)calloc(24, sizeof(char));
            if (NULL == loadReq.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 24);
                relaunch();
            }
            snprintf(loadReq.keyvalues[num].value, 24, "%f",
                     gptrMyHost->gpuinfo_dynamic[i].avail_gpu_mem);
            num ++;

            loadReq.keyvalues[num].key = REQUEST_KEY_GPU_STATUS;
            loadReq.keyvalues[num].value = copy_string(gptrMyHost->gpuinfo_dynamic[i].status);
            num++;

            loadReq.keyvalues[num].key = REQUEST_KEY_GPU_MODE;
            loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == loadReq.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(loadReq.keyvalues[num].value, 12, "%d", gptrMyHost->gpuinfo_dynamic[i].gpu_mode);
            num ++;

            loadReq.keyvalues[num].key = REQUEST_KEY_GPU_UT;
            loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == loadReq.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(loadReq.keyvalues[num].value, 12, "%f", gptrMyHost->gpuinfo_dynamic[i].gpu_ut);
            num ++;

            loadReq.keyvalues[num].key = REQUEST_KEY_GPU_TEMPERATURE;
            loadReq.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == loadReq.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(loadReq.keyvalues[num].value, 12, "%d",
                     gptrMyHost->gpuinfo_dynamic[i].gpu_temperature);
            num ++;
        }
    }
    loadReq.number = num;

    int size;
    size = gf_xdrsize_request(&loadReq);
    size += HEADER_LEN;

    char *request;
    request = (char *)calloc(size, sizeof(char));
    if (NULL == request) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        gf_xdr_free(gf_xdr_request, &loadReq);

        return;
    }

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    struct header tohdr;
    gf_init_header(&tohdr);
    tohdr.reqcode = JM_LOAD_UPD;

    if (!gf_xdr_message(&xdrs, &loadReq, &tohdr, NULL, gf_xdr_request, NULL)) {

        log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "gf_xdr_message(gf_xdr_request)");

        gf_xdr_free(gf_xdr_request, &loadReq);
        xdr_destroy(&xdrs);
        FREEUP(request);

        return;
    }

    struct header outhdr;
    char *reply;
    int cc, fd=-1;
    cc = gf_callmaster(gstrBossNode, g_jm_port, g_conntimeout, g_recvtimeout, 1,
                       XDR_GETPOS(&xdrs), request, &outhdr, &reply, &fd);

    gf_xdr_free(gf_xdr_request, &loadReq);
    FREEUP(request);
    xdr_destroy(&xdrs);
    if (0 < fd) {
        gf_chan_close(fd);
    }

    if (0 > cc) {

        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "gf_callmaster", g_masterlist);

        if (-2 == cc) { // master is down

            gptrMyCluster->number_inactivity ++;

            if (0 < gintJsFd) {
                gf_chan_close(gintJsFd);
                gintJsFd = -1;
                if (gboolMaster) {
                    gboolJSReady = FALSE;
                }
            }
            FREEUP(gstrBossNode);
        }

        return;
    }

    time_t bootTime = 0;
    if (0 == outhdr.reqcode) {

        // got reply from master, decode its boot time.
        xdrmem_create(&xdrs, reply, outhdr.length, XDR_DECODE);

        if (!xdr_long(&xdrs, (long *)&bootTime)) {
            log_message(ERR, NOCLASS, STR_FUNC_FAIL, __func__, "xdr_long");
            xdr_destroy(&xdrs);
            FREEUP(reply);

            return;
        }

        xdr_destroy(&xdrs);
    }
    FREEUP(reply);

    struct hostNode *host;
    host = (struct hostNode *)gf_hash_find(gptrMyCluster->node_table, gstrBossNode);
    if (NULL == host) {
        log_message(ERR, NOCLASS, "%s: The master node <%s> is unknown to me, shutdown myself.",
                    __func__, gstrBossNode);
        jm_die(DAEMON_QUIT_FATAL);
    }

    gptrMyCluster->number_inactivity = 0;
    gptrMyCluster->boss_node = host;

    if (bootTime != gtimeMasterJmBoot) { // master rebooted, resend my static info mation.
        if (0 <= send_static_info()) {
            gtimeMasterJmBoot = bootTime;
        }
    }

    return;
} // end function send_myload

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

/**
 * @brief       Send my node static information to master.
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 *
 * @note        This function should be called once after jm start.
 */
static int
send_static_info(void)
{
    if (NULL == gstrBossNode) { // do not know master for now.
        return -1;
    }

    struct header outhdr;
    gf_init_header(&outhdr);

    char buf[MAXMESSAGELENGTH/4];
    memset((char*)&buf, 0, sizeof(buf));

    XDR xdrs;
    xdrmem_create(&xdrs, buf, MAXMESSAGELENGTH/4, XDR_ENCODE);
    outhdr.reqcode = JM_STATIC_INFO;

    if (!gf_xdr_message(&xdrs, gptrMyHost, &outhdr, NULL, jm_xdr_staticinfo, NULL)) {
        log_message(ERR, NOCLASS, "%s: gf_xdr_message/jm_xdr_staticinfo failed.", __func__);
        return -1;
    }

    log_message(DEBUG, CHANNEL, "%s: sending static info to master <%s>", __func__, gstrBossNode);

    int cc, fd = -1;
    char *reply;
    cc = gf_callmaster(gstrBossNode, g_jm_port, g_conntimeout, g_recvtimeout, 1,
                       XDR_GETPOS(&xdrs), buf, &outhdr, &reply, &fd);

    if (0 > cc) { // something wrong with communication.
        log_message(ERR, NOCLASS, "%s: failed to send static info to master <%s>.",
                    __func__, g_masterlist);
        FREEUP(reply);
        return -1;
    }

    if (0 < fd) {
        epoll_ctl(gf_get_epfd(), EPOLL_CTL_DEL, fd, NULL);
        gf_chan_close(fd);
    }

    FREEUP(reply);
    xdr_destroy(&xdrs);

    if (0 != outhdr.reqcode) { // master reply error to us.
        return -1;
    }

    return 0;
} // end function send_static_info
